package com.alex.pull;

import com.alex.constants.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class PullConsumerStarter {

    // key为指定的队列, value为这个队列最后拉取数据的位置
    private static final Map<MessageQueue, Long> offsetTable = new HashMap<>();

    public static void main(String[] args) throws MQClientException {
        DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer("test_pull_consumer");

        pullConsumer.setNamesrvAddr(Const.NAMESRV_ADDR);

        pullConsumer.start();

        Set<MessageQueue> messageQueues = pullConsumer.fetchSubscribeMessageQueues("test-pull-topic");

        for (MessageQueue messageQueue : messageQueues) {
            System.out.println("consume from the queue" + messageQueue);

            SINGLE_MQ:
            while (true) {
                try {
                    PullResult pullResult = pullConsumer.pullBlockIfNotFound(messageQueue, null,
                            getMessageQueueOffset(messageQueue), 32);

                    System.out.println(pullResult);
                    System.out.println(pullResult.getPullStatus());
                    System.out.println();

                    PullStatus pullStatus = pullResult.getPullStatus();
                    putMessageQueueOffset(messageQueue, pullResult.getNextBeginOffset());
                    switch (pullStatus) {
                        case FOUND:
                            List<MessageExt> list = pullResult.getMsgFoundList();
                            for (MessageExt msg : list) {
                                System.out.println(new String(msg.getBody()));
                            }
                            break;
                        case NO_NEW_MSG:
                            System.out.println("没有消息了");
                            break SINGLE_MQ;
                        case NO_MATCHED_MSG:
                            break;
                        case OFFSET_ILLEGAL:
                            break;
                        default:
                            break;
                    }
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }

        pullConsumer.shutdown();
    }

    private static void putMessageQueueOffset(MessageQueue messageQueue, long nextBeginOffset) {
        offsetTable.put(messageQueue, nextBeginOffset);
    }

    private static long getMessageQueueOffset(MessageQueue messageQueue) {
        Long offset = offsetTable.get(messageQueue);
        if (offset != null) {
            return offset;
        }

        return 0;
    }
}
